use crate::cache::IndexChange;
use crate::keys::VddKey;
use crate::{cache, CF_LOG, ROCKSDB};
use chrono::Local;
use rust_rocksdb::WriteBatchWithTransaction;
use serde_json::json;
use std::sync::Arc;
use visit_dd_core::proj::VddMessage;
use visit_dd_core::{Ares, UserId, AC_ADD};

/// 将消息修改为已读状态
pub fn update_read_ts(
    batch: &mut WriteBatchWithTransaction<true>,
    receiver: &UserId,
    sender: &UserId,
) {
    let msg_s = cache::get_index(&VddKey::IndexUser2AppMessage(receiver.clone()).to_string());
    let cf_log = ROCKSDB.cf_handle(CF_LOG).unwrap();
    let keys = msg_s
        .iter()
        .map(|x| (cf_log, VddKey::EntityAppMessage(*x).to_string()));
    ROCKSDB.multi_get_cf(keys).iter().for_each(|item| {
        if let Ok(Some(vu8)) = item {
            let mut msg = serde_json::from_slice::<VddMessage>(vu8).unwrap();
            if msg.sender == sender.to_owned() && msg.read_ts.is_none() {
                msg.read_ts = Some(Local::now().naive_local());
                let k = VddKey::EntityAppMessage(msg.id).to_string();
                batch.put_cf(cf_log, k, json!(&msg).to_string());
            }
        }
    });
}
/// 获取一个用户的全部消息
pub fn select_some_by_uid(uid: &UserId) -> Vec<Arc<VddMessage>> {
    let mut msgs = Vec::<Arc<VddMessage>>::new();
    let idx = cache::get_index(&VddKey::IndexUser2AppMessage(uid.clone()).to_string());
    let cf_log = ROCKSDB.cf_handle(CF_LOG).unwrap();
    let keys = idx
        .iter()
        .map(|x| (cf_log, VddKey::EntityAppMessage(*x).to_string()));
    ROCKSDB.multi_get_cf(keys).iter().for_each(|item| {
        if let Ok(Some(vu8)) = item {
            let msg = serde_json::from_slice::<VddMessage>(vu8).unwrap();
            msgs.push(msg.into());
        }
    });
    msgs
}

/// 获取一个用户的未读消息计数
pub fn count_unread_by_uid(uid: &UserId) -> u64 {
    let mut count: u64 = 0;
    let idx = cache::get_index(&VddKey::IndexUser2AppMessage(uid.clone()).to_string());
    let cf_log = ROCKSDB.cf_handle(CF_LOG).unwrap();
    let keys = idx
        .iter()
        .map(|x| (cf_log, VddKey::EntityAppMessage(*x).to_string()));
    ROCKSDB.multi_get_cf(keys).iter().for_each(|item| {
        if let Ok(Some(vu8)) = item {
            let msg = serde_json::from_slice::<VddMessage>(vu8).unwrap();
            if msg.receiver == uid.to_owned() && msg.read_ts.is_none() {
                count += 1;
            }
        }
    });
    count
}

/// 数据层调用 保存一个消息
pub(crate) fn save(
    batch: &mut WriteBatchWithTransaction<true>,
    index_changes: &mut Vec<IndexChange>,
    msg: &VddMessage,
) {
    let k = VddKey::EntityAppMessage(msg.id).to_string();
    let cf_log = ROCKSDB.cf_handle(CF_LOG).unwrap();
    batch.put_cf(cf_log, k, json!(&msg).to_string());
    index_changes.push(IndexChange::new(
        VddKey::IndexUser2AppMessage(msg.receiver.clone()),
        AC_ADD,
        msg.id,
    ));
    // 给发送者的索引里也加一份，满足对话样式消息展示
    index_changes.push(IndexChange::new(
        VddKey::IndexUser2AppMessage(msg.sender.clone()),
        AC_ADD,
        msg.id,
    ));
}

/// api层使用的 保存一个消息
pub fn send(msg: &VddMessage) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    save(&mut batch, &mut index_changes, msg);
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?
    }
    ares
}
